package com.hjk.common.utils.concurrency;

import com.hjk.common.constants.GlobalConstant;
import com.hjk.common.utils.ThreadLocalMap;
import com.hjk.common.utils.check.Check;
import com.hjk.common.utils.collection.CollectUtils;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.junit.Test;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 多线程任务工具类
 *
 * @author huangjunkai
 * @version 1.0
 * @date 2019/10/15
 */
@Slf4j
@SuppressWarnings("unchecked")
public class FastTaskUtils {

    private static Runtime runtime = Runtime.getRuntime();
    /**
     * IO密集型任务使用 : 主要是进行IO操作，执行IO操作的时间较长，这是cpu出于空闲状态，导致cpu的利用率不高。
     */
    public final static int IO_CORE_POOL_SIZE = runtime.availableProcessors() * 2;
    /**
     * CPU密集型任务使用 :主要是执行计算任务，响应时间很快，cpu一直在运行，这种任务cpu的利用率很高。
     */
    public final static int CUP_CORE_POOL_SIZE = runtime.availableProcessors();





    /**
     *
     */
    //@Test
    public void test34232() {
        Thread thread = new Thread(() -> {
            int i = 1/0;
            System.out.println(111);
        },"exception");

        thread.setUncaughtExceptionHandler(new ExceptionHandler());

        getIoFastJobPool().execute(thread);
    }

    /**
     * 异常捕获处理器
     */
    public static class ExceptionHandler implements Thread.UncaughtExceptionHandler {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            log.error(t.getName() +":" + e.getMessage());

        }
    }


    public static void startTime() {
        ThreadLocalMap.put("countTime",System.nanoTime());
    }

    public static String endTime() {
        long start = (long)ThreadLocalMap.get("countTime");
        long end = System.nanoTime();
        String time = new BigDecimal((end - start) / 1000_000_000.0).setScale(6, RoundingMode.UP).toString();
        log.info("job time {} s", time);
        ThreadLocalMap.remove("countTime");
        return time;
    }

    /**
     * 方法计时工具类,获取方法运行时间
     *
     * @param runnable 任务
     */
    public static String getRunTime(Runnable runnable) {
        long start = System.nanoTime();
        runnable.run();
        long end = System.nanoTime();
        String time = new BigDecimal((end - start) / 1000_000_000.0).setScale(6, RoundingMode.UP).toString();
        log.info("job time {} s", time);
        return time;
    }

    /**
     * 方法计时工具类,获取方法运行时间
     *
     * @param runnable 任务
     */
    public static String getRunTime(String business,Runnable runnable) {
        long start = System.nanoTime();
        runnable.run();
        long end = System.nanoTime();
        String time = new BigDecimal((end - start) / 1000_000_000.0).setScale(6, RoundingMode.UP).toString();
        log.info("<<<<<<<<<<[{}] job time {} s",business, time);
        return time;
    }


    /**
     * 方法计时工具类,获取方法运行时间
     *
     * @param supplier 任务
     */
    public static Object getRunTime(String business, Supplier supplier) {
        long start = System.nanoTime();
        Object o = supplier.get();
        long end = System.nanoTime();
        String time = new BigDecimal((end - start) / 1000_000_000.0).setScale(6, RoundingMode.UP).toString();
        log.info("<<<<<<<<<<[{}] job time {} s",business, time);
        return o;
    }

    //@Test
    public void test3() throws InterruptedException {
        FastTaskUtils.getRunTime(() -> {
            int count = 100;
            String[] objects = new String[count];
            //CountDownLatch countDownLatch = FastTaskUtils.getCountDownLatch(count);
            CyclicBarrier cyclicBarrier = FastTaskUtils.getCyclicBarrier(count, () -> {
                log.info(Arrays.toString(objects));
            });
            //使用线程安全的集合
            //List<String> objects = Collections.synchronizedList( CollectUtils.newList(count));

            for (int i = 0; i < count; i++) {
                String num = i + "";
                int finalI = i;
                getIoFastJobPool("test3()",WorkQueue.synchronousQueue).execute(() -> {
                     /* try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }*/
                    objects[finalI] = num;
                    //countDownLatch.countDown();

                    if ("99".equals(num)) {
                        try {
                            TimeUnit.SECONDS.sleep(5000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        cyclicBarrier.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                });
            }

           /* try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/

        });


    }

    /**
     * 执行异步任务 无返回值
     *
     * @param runnable 任务
     * @param executor 自定义线程池
     */
    public static void runAsyncTask(Runnable runnable, Executor executor) {
        CompletableFuture.runAsync(runnable, executor);
    }

    /**
     * 执行异步任务 有返回值
     *
     * @param supplier 任务
     * @param executor 自定义线程池
     */
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
        return  CompletableFuture.supplyAsync(supplier, executor);
    }

    /**
     * 执行异步任务 有返回值
     *
     * @param supplier 任务
     */
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return  CompletableFuture.supplyAsync(supplier);
    }

    /**
     * 执行异步任务
     *
     * @param runnable 任务
     */
    public static void runAsyncTask(Runnable runnable) {
        CompletableFuture.runAsync(runnable);
    }


    /**
     * 获取计数器
     *
     * @param count 次数
     * @return CountDownLatch
     */
    public static CountDownLatch getCountDownLatch(int count) {
        return new CountDownLatch(count);
    }

    /**
     * 获取循环屏障
     *
     * @param parties 次数
     * @return {@link CyclicBarrier}) CyclicBarrier
     */
    public static CyclicBarrier getCyclicBarrier(int parties) {
        return new CyclicBarrier(parties);
    }

    /**
     * 获取循环屏障
     *
     * @param parties  次数
     * @param runnable 最终执行方法
     * @return {@link CyclicBarrier}) CyclicBarrier
     */
    public static CyclicBarrier getCyclicBarrier(int parties, Runnable runnable) {
        return new CyclicBarrier(parties, runnable);
    }


    /**
     * 创建固定线程
     * @param nThreads
     * @return
     */
    public static ExecutorService getNewFixedThreadPool(int nThreads,String threadNamePrefix) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new NameTreadFactory(threadNamePrefix),
                new ThreadPoolExecutor.AbortPolicy());
    }



    /**
     * 快速任务线程池,执行IO密集型任务
     */
    public static ThreadPoolExecutor getNewFixedThreadIOPool(String threadNamePrefix) {
        return new ThreadPoolExecutor(
                IO_CORE_POOL_SIZE,
                IO_CORE_POOL_SIZE,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new NameTreadFactory(threadNamePrefix),
                new MyPolicy()
        );
    }



    public static ThreadPoolExecutor getIoFastJobPool() {
        return getIoFastJobPool("ioFastJobPool", WorkQueue.LinkedBlockingQueue);
    }

    /**
     * 快速任务线程池,执行IO密集型任务
     */
    public static ThreadPoolExecutor getIoFastJobPool(String threadNamePrefix, WorkQueue workQueue) {
        BlockingQueue<Runnable> queue = new SynchronousQueue<>();
        if (WorkQueue.LinkedBlockingQueue.equals(workQueue)) {
            queue = new LinkedBlockingQueue<>();
        } else if (WorkQueue.synchronousQueue.equals(workQueue)) {
            queue = new SynchronousQueue<>();
        }
        return new ThreadPoolExecutor(
                IO_CORE_POOL_SIZE,
                Integer.MAX_VALUE,
                60L,
                TimeUnit.SECONDS,
                queue,
                new NameTreadFactory(threadNamePrefix),
                new MyPolicy()
        );
    }

    @Getter
    @NoArgsConstructor
    @AllArgsConstructor
    enum WorkQueue {
        /**
         * 同步队列
         */
        synchronousQueue,
        /**
         * 链表队列
         */
        LinkedBlockingQueue;

        String type;
    }

    /**
     * 快速任务线程池,执行CPU密集型任务,不考虑oom
     */
    public static ThreadPoolExecutor getCpuFastJobPool() {
        return new ThreadPoolExecutor(
                CUP_CORE_POOL_SIZE,
                Integer.MAX_VALUE,
                60L,
                TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                new NameTreadFactory(),
                new MyPolicy()
        );
    }

    /**
     * 自定义线程工厂,为线程取个名字
     */
    static class NameTreadFactory implements ThreadFactory {
        /**
         * 线程名称前缀
         */
        private String threadNamePrefix;

        private final AtomicInteger mThreadNum = new AtomicInteger(0);

        public NameTreadFactory() {
        }

        public NameTreadFactory(String threadNamePrefix) {
            this.threadNamePrefix = threadNamePrefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            return ObjectUtils.isEmpty(threadNamePrefix) ? new Thread(r, "fastTask-thread-" + mThreadNum.getAndIncrement()) : new Thread(r, threadNamePrefix + "-" + mThreadNum.getAndIncrement());
        }
    }

    /**
     * 线程队列满了,并且达到最大线程数 之后执行的拒接策略 Policy
     * {@link ThreadPoolExecutor.AbortPolicy}): 丢弃任务并抛出ExecutioRejectednException异常。 (默认)
     * {@link ThreadPoolExecutor.DiscardPolicy} ：也是丢弃任务，但是不抛出异常。
     * {@link ThreadPoolExecutor.DiscardOldestPolicy} ：丢弃队列最前面的任务，然后重新尝试执行任务。（重复此过程）
     * {@link ThreadPoolExecutor.CallerRunsPolicy}：由调用线程处理该任务。
     */
    public static class MyPolicy extends ThreadPoolExecutor.CallerRunsPolicy {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            Thread r1 = (Thread) r;
            // 可做日志记录等
            log.warn("线程池已满! 调用线程处理该任务 : {}! ", r1.getName());
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }


    /**
     * 多线程任务await 无返回值
     * taskList 任务list
     *
     * @param taskList
     */
    public void awaitNoResult(List<CompletableFuture<Void>> taskList) {
        if (ObjectUtils.isNotEmpty(taskList)) {
            //线程阻塞
            CompletableFuture.allOf(taskList.toArray(new CompletableFuture[0])).join();
        }

    }

    //@Test
    public void test345334() {
        CompletableFuture<Integer> a = CompletableFuture.supplyAsync(() -> {
            return 1;
        });
        CompletableFuture<Integer> b = CompletableFuture.supplyAsync(() -> {
            return 2;
        });
        CompletableFuture<Integer> c = CompletableFuture.supplyAsync(() -> {
            return 3;
        });

        List<Integer> list = awaitHasResult(Arrays.asList(a, b, c));
        System.out.println(list);
    }





    /**
     * 多线程任务await 有返回值
     *
     * @param taskList 任务list
     * @param <T>      结果泛型
     * @return
     */
    @SneakyThrows
    public static <T> List<T> awaitHasResult(List<CompletableFuture<T>> taskList) {
        return ObjectUtils.isNotEmpty(taskList) ?
                CompletableFuture.allOf(taskList.toArray(new CompletableFuture[0]))
                        .thenApply(v -> taskList.stream().map(CompletableFuture::join)
                                .collect(Collectors.toList())).get() : null;

    }


}
